package com.apex.flink.source;

import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.connector.jdbc.dialect.JdbcDialect;
import org.apache.flink.connector.jdbc.internal.options.JdbcLookupOptions;
import org.apache.flink.connector.jdbc.internal.options.JdbcOptions;
import org.apache.flink.connector.jdbc.table.JdbcLookupFunction;
import org.apache.flink.connector.jdbc.table.JdbcRowDataInputFormat;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.connector.source.InputFormatProvider;
import org.apache.flink.table.connector.source.ScanTableSource;
import org.apache.flink.table.functions.AsyncTableFunction;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.table.sources.LookupableTableSource;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.types.Row;

import static org.apache.flink.table.types.utils.TypeConversions.fromDataTypeToLegacyInfo;


public class OracleDynamicTableSource implements ScanTableSource, LookupableTableSource<Row> {
    private final TableSchema tableSchema;
    private final JdbcOptions options;
    private final DataType producedDataType;
    private final JdbcLookupOptions lookupOptions;

    public OracleDynamicTableSource(TableSchema tableSchema, JdbcOptions options, DataType producedDataType,
                                    JdbcLookupOptions lookupOptions) {
        this.tableSchema = tableSchema;
        this.options = options;
        this.producedDataType = producedDataType;
        this.lookupOptions = lookupOptions;
    }

    @Override
    public TableFunction<Row> getLookupFunction(String[] lookupKeys) {
        final RowTypeInfo rowTypeInfo = (RowTypeInfo) fromDataTypeToLegacyInfo(producedDataType);
        return JdbcLookupFunction.builder()
                .setOptions(options)
                .setLookupOptions(lookupOptions)
                .setFieldTypes(rowTypeInfo.getFieldTypes())
                .setFieldNames(rowTypeInfo.getFieldNames())
                .setKeyNames(lookupKeys)
                .build();
    }

    @Override
    public AsyncTableFunction<Row> getAsyncLookupFunction(String[] lookupKeys) {
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean isAsyncEnabled() {
        return false;
    }

    @Override
    public ChangelogMode getChangelogMode() {
//        return ChangelogMode
//                .newBuilder()
//                .addContainedKind(RowKind.INSERT)
//                .addContainedKind(RowKind.DELETE)
//                .build();
        return ChangelogMode.insertOnly();
    }

    @Override
    public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) {
        final JdbcDialect dialect = options.getDialect();

        String query = dialect.getSelectFromStatement(
                options.getTableName(), tableSchema.getFieldNames(), new String[0])
                .replaceAll("`"," ");

        final RowType rowType = (RowType) tableSchema.toRowDataType().getLogicalType();

        final JdbcRowDataInputFormat.Builder builder = JdbcRowDataInputFormat.builder()
                .setDrivername(options.getDriverName())
                .setDBUrl(options.getDbURL())
                .setUsername(options.getUsername().get())
                .setPassword(options.getPassword().get())
                .setQuery(query)
                .setRowConverter(dialect.getRowConverter(rowType))
                .setRowDataTypeInfo((runtimeProviderContext
                        .createTypeInformation(tableSchema.toRowDataType())));

        return InputFormatProvider.of(builder.build());
    }

    @Override
    public DynamicTableSource copy() {
        return new OracleDynamicTableSource(tableSchema,options, producedDataType, lookupOptions);
    }

    @Override
    public String asSummaryString() {
        return "oracle";
    }


    @Override
    public TableSchema getTableSchema() {
        return null;
    }
}
